package com.atguigu.bigdata.spark.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Duration, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

//使用9999端口，发送数据,来进行批次处理
object SprakStreaming01_WordCount {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamWordCount")

    //构建流式数据的上下文环境对象，并且设置处理周期
    val ssc = new StreamingContext(conf,Duration(3000))

    //从网络端口获取数据,拿到一行字符串
    val socketData: ReceiverInputDStream[String] = ssc.socketTextStream("localhost",9999)
    //将数据进行扁平化处理
    val words: DStream[String] = socketData.flatMap(_.split(" "))
    //将数据结构进行转换
    val wordToOne: DStream[(String, Int)] = words.map((_,1))
    //聚合操作(value1+value2)
    val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)
    wordToCount.print()



    //启动数据采集器,采集不可停止
    ssc.start()
    //主线程需要阻塞,否则main完成则释放。
    ssc.awaitTermination()






   // ssc.stop()
  }

}
